package io.jmqtt.server;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.jmqtt.handler.MqttConnectHandler;
import io.jmqtt.handler.MqttMessageHandler;
import io.jmqtt.handler.Pipelines;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;

/**
 * JMQTT 启动类
 * 
 * @author chenhua
 *
 */
public class JMQTTServer {

	private static final Logger LOG = LoggerFactory.getLogger(JMQTTServer.class);

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		LOG.info("Starting JMQTT Server");

		JMQTTServer.start(1883);
	}

	public static void start(int port) throws InterruptedException {
		long startTime = System.nanoTime();
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		try {
			ServerBootstrap b = new ServerBootstrap();
			b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
					.childHandler(new ChannelInitializer<SocketChannel>() {
						@Override
						public void initChannel(SocketChannel ch) throws Exception {
							ch.pipeline().addLast(Pipelines.MQTT_MESSAGE_DECODER, new MqttDecoder());
							ch.pipeline().addLast(Pipelines.MQTT_MESSAGE_ENCODER, MqttEncoder.INSTANCE);
							ch.pipeline().addLast(Pipelines.MQTT_CONNECT, new MqttConnectHandler());
							ch.pipeline().addLast(Pipelines.MQTT_MESSAGE_HANDLER, new MqttMessageHandler());

						}
					});

			ChannelFuture f = b.bind(port).sync();
			LOG.info("Started JMQTT at port {} in {}ms", port,
					TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime));
			f.channel().closeFuture().sync();
		} catch (InterruptedException e) {
			LOG.error("Started error", e);
		} finally {
			workerGroup.shutdownGracefully();
			bossGroup.shutdownGracefully();
		}
	}

}
